package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.hmdp.dto.Result;

import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.RedisCache;
import com.hmdp.utils.RedisWork;
import com.hmdp.utils.UserHolder;

import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

import static com.hmdp.utils.RedisConstants.*;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Resource
    private ISeckillVoucherService seckillVoucherService;
    @Autowired
    private VoucherOrderServiceImpl self;
    @Resource
    private RedisWork redisWork;
    @Resource
    private RedisCache redisCache;
    @Resource
    private RedissonClient redissonClient;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 抢购秒杀卷
     *
     * @param voucherId
     * @return
     */
    //连接lua脚本
    public static final DefaultRedisScript<Long> secKillScript;
    static {
        secKillScript = new DefaultRedisScript<>();
        secKillScript.setLocation(new ClassPathResource("seckill.lua"));
        secKillScript.setResultType(Long.class);
    }

    //异步处理线程池
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();

    //在类初始化之后执行，因为当这个类初始化好了之后，随时都是有可能要执行的
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    private String streamKey = "stream.orders";

    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    //1.获取消息队列的任务 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    List<MapRecord<String, Object, Object>> mapRecords =
                            stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1L).block(Duration.ofSeconds(2)),
                            StreamOffset.create(streamKey, ReadOffset.lastConsumed())
                    );
                    //2.获取消息失败，代表没有消息，进入下一轮循环
                    if(mapRecords == null || mapRecords.isEmpty()){
                        try {
                            Thread.sleep(50);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        continue;
                    }

                    //3.获取消息成功，处理消息
                    MapRecord<String, Object, Object> record = mapRecords.get(0);
                    Map<Object, Object> value = record.getValue();//value中为键值对
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                    //4.将处理好的消息传入方法，开始下单
                    handVoucherOrder(voucherOrder);
                    //5.ack确认
                    stringRedisTemplate.opsForStream().acknowledge(streamKey,"g1",record.getId());
                } catch (RuntimeException e) {
                    log.error("处理订单异常");
                    hangPendingList();
                }
            }
        }
        //处理pending-list的已消费未确认的消息
        private void hangPendingList(){
            while (true) {
                try {
                    //1.获取pending-list的消息 XREADGROUP GROUP g1 c1 COUNT 1 STREAMS stream.orders 0
                    List<MapRecord<String, Object, Object>> mapRecords =
                            stringRedisTemplate.opsForStream().read(
                                    Consumer.from("g1", "c1"),
                                    StreamReadOptions.empty().count(1L),
                                    StreamOffset.create(streamKey, ReadOffset.from("0"))
                            );
                    //2.获取消息失败，代表pending-list没有异常消息
                    if(mapRecords == null || mapRecords.isEmpty()){
                        break;
                    }

                    //3.获取消息成功，处理消息
                    MapRecord<String, Object, Object> record = mapRecords.get(0);
                    Map<Object, Object> value = record.getValue();//value中为键值对
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);

                    //4.将处理好的消息传入方法，开始下单
                    handVoucherOrder(voucherOrder);
                    //5.ack确认
                    stringRedisTemplate.opsForStream().acknowledge(streamKey,"g1",record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常错误");
                    try {
                        Thread.sleep(50);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            }
        }

    }

    private void handVoucherOrder(VoucherOrder voucherOrder) {
        //获取锁对象---使用redisson
        RLock lock = redissonClient.getLock("lock:order:" + voucherOrder.getUserId());
        //尝试获取锁----最大等待时间（默认-1，即失败立即返回）、锁失效时间（默认30s）、时间单位
        boolean success = lock.tryLock();

        if(!success){
            //获取锁失败
            log.error("不能重复下单单单单单单单单单");
            return;
        }
        try {
            self.createVoucherOrder(voucherOrder);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        //一人只能购买一单
        Long voucherId = voucherOrder.getVoucherId();
        Integer count = query()
                .eq("voucher_id", voucherId)
                .eq("user_id", voucherOrder.getUserId()).count();

        if(count > 0){
            //该用户已买秒杀卷
            log.error("您已经购买了秒杀卷");
            return ;
        }

        //扣减库存--乐观锁
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .gt("stock",0)
                .update();
        if(!success){
            log.error("库存不足");
            return ;
        }

        //订单id
        save(voucherOrder);
    }

    /**
     * 利用异步操作完成秒杀优化，因为
     * 1.查询优惠卷
     * 2.判断秒杀库存是否足够
     * 3.查询订单
     * 4.校验是否是一人一单
     * 5.扣减库存
     * 6.创建订单
     * 在这六步操作中，有很多操作要操作数据库，且是一个线程串行执行， 这会导致我们的程序执行很慢
     * 解决方案：
     * 将 2.判断秒杀库存是否足够、4.校验是否是一人一单 交给redis判断（可用lua，保证原子性），
     * 这样我们只需要进行快速的逻辑判断，根本就不用等下单逻辑走完，我们直接给用户返回成功
     * 再在后台开一个线程，后台线程慢慢地去执行queue里边的消息
     */
    public Result getkillByvoucherId(Long voucherId) {
        //设置KEYS[]
        String secKillKey = SECKILL_STOCK_KEY + voucherId;
        String orderKey = SECKILL_ORDER_KEY + voucherId;
        ArrayList<String> list = new ArrayList<>();
        //设置ARGV[]
        Long userId = UserHolder.getUser().getId();
        long orderId = redisWork.getId("order");

        //执行lua脚本
        Collections.addAll(list, secKillKey, orderKey);
        Long result = stringRedisTemplate.execute(
                secKillScript,
                list,
                userId.toString(), String.valueOf(orderId), voucherId.toString()
        );
        int r = result.intValue();
        if (r == 1) {
            return Result.fail("库存不够");
        } else if (r == 2) {
            return Result.fail("您已经下单");
        }
        //result == 0
        return Result.ok(orderId);
    }

    /*public Result getkillByvoucherId(Long voucherId) {
        //获取秒杀表的秒杀卷
        SeckillVoucher seckillVoucher = seckillVoucherService.getById(voucherId);
        //秒杀卷是否开始
        if (seckillVoucher.getBeginTime().isAfter(LocalDateTime.now())) {
            return Result.fail("秒杀卷尚未开始");
        }
        //秒杀卷是否结束
        if (seckillVoucher.getEndTime().isBefore(LocalDateTime.now())) {
            return Result.fail("秒杀卷已经结束");
        }
        //秒杀卷是否库存充足
        if(seckillVoucher.getStock() <= 0){
            return Result.fail("库存bu足");
        }

        Long userId = UserHolder.getUser().getId();
        *//**
     * 不能在函数内部加锁：
     *      因为锁释放后，函数未结束（提交），这时又有线程进入锁中，查到订单表中没数据（因为没提交），往表里插数据
     * 不要用 this.createVoucherOrder(voucherId) 调用事务
     *      因为调用事务方法要用动态代理对象，事务就是依靠动态代理实现的
     *      可用自引用调用
     * 当在一个Bean上调用一个方法时，如果这个方法被@Transactional注解，那么Spring会创建一个代理对象来包装原始对象
     *//*
     *//*synchronized (userId.toString().intern()){
            return self.createVoucherOrder(voucherId);
        }*//*

        //使用redis锁来实现一人一单问题
        //使用自己写的锁
        //boolean success = redisCache.addLock("lock:order:" + userId, LOCK_ORDER_TTL);

        //获取锁对象---使用redisson
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //尝试获取锁----最大等待时间（默认-1，即失败立即返回）、锁失效时间（默认30s）、时间单位
        boolean success = lock.tryLock();

        if(!success){
            //获取锁失败
            return Result.fail("一人只能买一单！");
        }
        try {
            return self.createVoucherOrder(voucherId);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            redisCache.releaseLock("lock:order:" + userId);
        }
    }*/

    /**
     * 创建订单
     * @param voucherId
     * @return
     */
    /*@Transactional
    public Result createVoucherOrder(Long voucherId) {
        //一人只能购买一单
        Long userId = UserHolder.getUser().getId();

        Integer count = query()
                .eq("voucher_id", voucherId).eq("user_id", userId).count();
        if(count > 0){
            //该用户已买秒杀卷
            return Result.fail("您已经购买了秒杀卷");
        }

        //扣减库存--乐观锁
        boolean success = seckillVoucherService.update()
                .setSql("stock = stock - 1")
                .eq("voucher_id", voucherId)
                .gt("stock",0)
                .update();
        if(!success) return Result.fail("库存不足");

        //创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        //订单id
        long orderId = redisWork.getId("order");
        voucherOrder.setId(orderId);

        voucherOrder.setVoucherId(voucherId);//代金卷id

        voucherOrder.setUserId(userId);//用户id

        save(voucherOrder);

        return Result.ok(orderId);
    }*/
}
